
package com.iot08.push.kafka;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.iot08.common.constant.Const;
import com.iot08.common.constant.ConstKafka;
import com.iot08.common.constant.ConstPush;
import com.iot08.common.entity.app.*;
import com.iot08.common.utils.CollectionUtil;
import com.iot08.common.utils.HttpReq;
import com.iot08.common.utils.StringUtil;
import com.iot08.push.mqtt.handle.IMqttSender;
import com.iot08.push.service.*;
import com.iot08.push.utils.*;
import com.turo.pushy.apns.ApnsClient;
import com.turo.pushy.apns.ApnsClientBuilder;
import com.turo.pushy.apns.PushNotificationResponse;
import com.turo.pushy.apns.util.ApnsPayloadBuilder;
import com.turo.pushy.apns.util.SimpleApnsPushNotification;
import com.turo.pushy.apns.util.concurrent.PushNotificationFuture;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import okhttp3.Credentials;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.codec.binary.StringUtils;
import org.apache.commons.io.FileUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.bson.Document;
import org.bson.types.ObjectId;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.io.File;
import java.io.IOException;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;


/**
 * @version 1.0
 * @Description: kafka消费者
 * @time: 2020/9/21 14:26
 * @author: Rock
 */

@Slf4j
@Component
public class KafkaConsumer {

    @Resource
    private ChatGroupService chatGroupService;

    @Resource
    private SingleChatGroupService singleChatGroupService;

    @Resource
    private AccountService accountService;
    @Resource
    private DeviceTokenService deviceTokenService;
    @Resource
    private IMqttSender iMqttSender;
    @Resource
    private OpenidConfigService openidConfigService;
    @Resource
    private AsyncService asyncService;

    @Resource
    private VipMemberService vipMemberService;

    @Value("${mqtt.server}")
    private String mqttServer;

    @Value("${mqtt.http.apiKey:}")
    private String mqttHttpApiKey;

    @Value("${mqtt.http.secretKey:}")
    private String mqttHttpSecretKey;

    @Value("${spring.kafka.consumer.topic-name}")
    private String consumerTopicName;

    //不消费kafka消息,用于消息堵塞时快速消费消息的操作
    @Value("${notConsumerKafkaMsg:0}")
    private String notConsumerKafkaMsg;


    //    private static final String VOIPDIS = "&voipdis";
//    private static final String VOIPDEV = "&voipdev";
    private static final String DIS = "&dis";
    private static final String DEV = "&dev";
    private static Map<String, ApnsClient> pushClients = new ConcurrentHashMap<String, ApnsClient>();

    static {
        JSONObject pushConfig = null;
        //获取配置
        try {
           /* String input = FileUtils.readFileToString(new File("./cert/apns.json"), "UTF-8");
            pushConfig = JSONObject.parseObject(input);*/
        } catch (Exception error) {
            log.info("初始化PushConfig失败: " + error.getMessage());
            log.error("初始化PushConfig失败: " + error.getMessage(), error);
        }
        if (pushConfig != null) {
            for (String itemKey : pushConfig.keySet()) {
                JSONObject item = pushConfig.getJSONObject(itemKey);
                String pwd = item.getString("password");
                String devPath = item.getString("develop");
                String disPath = item.getString("produce");
//            String voipPath = item.getString("voip");
                try {
//                //voip_dis
//                if (voipPath != null && voipPath.length() > 0) {
//                    ApnsClient voip_dis_apnsClient = new ApnsClientBuilder()
//                            .setEventLoopGroup(new NioEventLoopGroup(2))
//                            .setConcurrentConnections(4)
//                            .setApnsServer(ApnsClientBuilder.PRODUCTION_APNS_HOST, ApnsClientBuilder.ALTERNATE_APNS_PORT)
//                            .setClientCredentials(new File(voipPath), pwd)
//                            .build();
//                    pushClients.put(itemKey + VOIPDIS, voip_dis_apnsClient);
//                    //voip_dev
//                    ApnsClient voip_dev_apnsClient = new ApnsClientBuilder()
//                            .setApnsServer(ApnsClientBuilder.DEVELOPMENT_APNS_HOST, ApnsClientBuilder.ALTERNATE_APNS_PORT)
//                            .setClientCredentials(new File(voipPath), pwd)
//                            .build();
//                    pushClients.put(itemKey + VOIPDEV, voip_dev_apnsClient);
//                }

                    //dis
                    ApnsClient dis_apnsClient = new ApnsClientBuilder()
                            .setEventLoopGroup(new NioEventLoopGroup(2))
                            .setConcurrentConnections(4)
                            .setApnsServer(ApnsClientBuilder.PRODUCTION_APNS_HOST, ApnsClientBuilder.ALTERNATE_APNS_PORT)
                            .setClientCredentials(new File(disPath), pwd)
                            .build();
                    pushClients.put(itemKey + DIS, dis_apnsClient);
                    //dev
                    ApnsClient dev_apnsClient = new ApnsClientBuilder()
                            .setApnsServer(ApnsClientBuilder.DEVELOPMENT_APNS_HOST, ApnsClientBuilder.ALTERNATE_APNS_PORT)
                            .setClientCredentials(new File(devPath), pwd)
                            .build();
                    pushClients.put(itemKey + DEV, dev_apnsClient);
                } catch (IOException e) {
                    log.info("pushClients 初始化异常！！");
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * @param
     * @time: 2020/9/23 17:07
     * @author: Rock
     * @version 1.0
     */
    @KafkaListener(groupId = "${spring.kafka.consumer.group-id}", topics = "${spring.kafka.consumer.topic-name}", concurrency = "${spring.kafka.consumer.concurrency}")
    public void listen(List<ConsumerRecord> messages) {
        if (messages != null) {
            for (ConsumerRecord message : messages) {
                String value = message.value().toString().replace("\r", "").replace("\n", "")
                        .replace("\ufeff", "");
                log.info("topic={}. key={}. value={}. partition={} notConsumerKafkaMsg={}",
                        message.topic(), message.key(), value, message.partition(), notConsumerKafkaMsg);
                if ("1".equals(notConsumerKafkaMsg)) {
                    log.info("不消费kafka消息,暂时忽略消息");
                    continue;
                }
                handerPushMsg(value);
            }
        } else {
            log.error("listen messages is null.");
        }
    }

    /**
     * 处理来着kafka的 回调消息
     *
     * @param message
     */
    public void handerPushMsg(String message) {
        try {
            JSONObject json = JSON.parseObject(message);
            if (json != null) {
                JSONObject data = json.getJSONObject(ConstPush.MSG_DATA);
                if (StringUtil.isBlank(data)) {
                    log.info("msgData is null. consumerTopicName={}, message={}", consumerTopicName, message);
                    return;
                }
                JSONObject dataCopy = JSONObject.parseObject(data.toJSONString());
                //log.info("json message={}", json);
                List<ApnsVO> apns = new ArrayList<>();
                List<MobVO> mobs = new ArrayList<>();
                //推送类型；1=圈子推送；2=多用户推送；3=单用户推送；4=只推送mqtt
                int pushType = 0;
                if (json.containsKey(ConstPush.PUSH_TYPE)) {
                    pushType = json.getInteger(ConstPush.PUSH_TYPE);
                }
                if (pushType == ConstPush.PUSHTYPE_1) {
                    log.info("GROUPID = {}", json.getString(ConstPush.GROUPID));
                    chatGroupHandle(json.getString(ConstPush.GROUPID), data, apns, mobs);
                } else if (pushType == ConstPush.PUSHTYPE_2) {
                    Object obj = json.get(ConstPush.OPENID_LIST);
                    log.info("OPENID_LIST = {}", obj);
                    if (obj instanceof List) {
                        multipleUsersHandle((List<String>) obj, data, apns, mobs);
                    }
                } else if (pushType == ConstPush.PUSHTYPE_3) {
                    log.info("OPENID = {}", json.getString(ConstPush.OPENID));
                    singleUsersHandle(json.getString(ConstPush.OPENID), data, apns, mobs);
                } else if (pushType == ConstPush.PUSHTYPE_4) {
                    //直接发送mqtt
                    singleUsersOnlySendMqttHandle(json.getString(ConstPush.TOPIC), data);
                    return;
                } else {
                    log.error("Message Exception :" + message);
                    return;
                }
                boolean isVideoChat = false;
                Map<String, String> extraMap = new HashMap<>();
                PushSendVO vo = new PushSendVO();
                JSONObject mes = data.getJSONObject(ConstPush.MESSAGE);
                JSONObject mesCopy = dataCopy.getJSONObject(ConstPush.MESSAGE);
                if (StringUtil.isNotBlank(mes)) {
                    if (ConstPush.ACCOUNT_MESSAGE.equals(data.getString(ConstPush.TYPE))) {
                        String dataType = mes.getString(ConstPush.TYPE);
                        String name = mes.getString(ConstPush.NAME) == null ? "" : mes.getString(ConstPush.NAME);
                        switch (dataType) {
                            case "photograph":
                                vo.setAlert(name + "上传照片");
                                vo.setMobType(1);
                                extraMap.put("currentImei", mes.getString(ConstPush.IMEI));
                                mesCopy.remove("photographID");
                                mesCopy.remove("thumbURL");
                                mesCopy.remove("url");
                                break;
                            case "LoginOffLine":
                                vo.setAlert("您的账号已在其他设备登录");
                                vo.setType("LoginOffLine");
                                break;
                            case "systemUpdate":
                                vo.setAlert(mes.getString("content"));
                                break;
                            case "rail":
                                if ("enter".equals(mes.getString("action"))) {
                                    vo.setAlert(name + "进入安全区域" + mes.getString("railName"));
                                } else {
                                    vo.setAlert(name + "离开安全区域" + mes.getString("railName"));
                                }
                                break;
                            case "sos":
                                vo.setAlert(name + "触发SOS");
                                break;
                            case "lowBattery":
                                vo.setAlert(name + "设备低电量");
                                break;
                            case "sms":
                                vo.setAlert(name + "收到新的代收短信");
                                break;
                            case "inquireChargeSms":
                                vo.setAlert(name + "收到查询话费流量的短信");
                                break;
                            case "wifiSearch":
                                vo.setAlert(name + "wifi搜索信息已上报");
                                break;
                            case "wifiCur":
                                vo.setAlert(name + "设备当前连接的wifi信息");
                                break;
                            case "wifiSet":
                                vo.setAlert(name + "设备wifi设置的应答消息");
                                break;
                            case "callLog":
                                vo.setAlert(name + "设备通话记录上报");
                                break;
                            case "applicationRrecord":
                                vo.setAlert(String.format("用户（%s）申请关注[%s] 手表", mes.getString("phone"), mes.getString("name")));
                                break;
                            case "weChatMsg":
                                vo.setAlert(name + "设备微聊已达上限");
                                break;
                            case "applicationOvertime":
                                if (mes.getInteger("own") == 1) {
                                    vo.setAlert(String.format("你申请关注【%s】手表，管理员未及时处理，申请已失效", mes.getString("name")));
                                } else {
                                    vo.setAlert(String.format("用户（%s）申请关注[%s] 手表, 记录未及时处理，已失效", mes.getString("phone"), mes.getString("name")));
                                }
                                break;
                            case "checkedResults":
                                if ("agree".contentEquals(mes.getString("action"))) {
                                    vo.setAlert(String.format("手表（%s）的管理员已同意您 关注手表", mes.getString("imei")));
                                } else {
                                    vo.setAlert(String.format("手表（%s）的管理员已拒绝您 关注手表", mes.getString("imei")));
                                }
                                break;
                            case "checkResults":
                                if ("agree".contentEquals(mes.getString("action"))) {
                                    vo.setAlert(String.format("用户（%s）申请关注[%s] 手表, 您已同意此用户关注手表", mes.getString("phone"), mes.getString("name")));
                                } else {
                                    vo.setAlert(String.format("用户（%s）申请关注[%s] 手表, 您已拒绝此用户关注手表", mes.getString("phone"), mes.getString("name")));
                                }
                                break;
                            case "identityTransfer":
                                vo.setAlert(String.format("您已把手表(%s)的管理员身份转移给[%s](%s)", mes.getString("imei"), mes.getString("name"), mes.getString("phone")));
                                break;
                            case "identityTransfered":
                                vo.setAlert(String.format("[%s]手表(%s)的管理员把 管理员身份转移给您", mes.getString("name"), mes.getString("imei")));
                                break;
                            case "CostFlow":
                                vo.setAlert(String.format("收到 %s 的话费流量查询短信", mes.getString("name")));
                                break;
                            case "powerModel":
                                String msg = "一个通知";
                                Integer status = mes.getInteger(ConstPush.STATUS);
                                if (Const.INT_0.equals(status)) {
                                    msg = String.format("【%s】已解除省电状态", name);
                                } else if (Const.INT_1.equals(status)) {
                                    msg = String.format("【%s】设备电量低，进入省电状态", name);
                                }
                                vo.setAlert(msg);
                                break;
                            case "reject":
                                isVideoChat = true;
                                vo.setVideoId(mes.getString("video_id"));
                                vo.setVideo_identify(mes.getString("video_identify"));
                                vo.setContentAvailable(true);
                                vo.setOpenid(mes.getString(ConstPush.OPENID));
                                vo.setAlert(String.format("设备：%s拒绝了通话", mes.getString(ConstPush.IMEI)));
                                break;
                            case "video":
                                isVideoChat = true;
                                vo.setContentAvailable(true);
                                vo.setVideo(1);
                                vo.setOpenid(mes.getString(ConstPush.OPENID));
                                vo.setAlert(String.format("%s邀请您通话", mes.getString(ConstPush.IMEI)));
                                vo.setMobType(2);
//                                String imei = mes.getString(ConstPush.IMEI);
//                                if (imei != null) {
//                                    if (imei.contains("imei-")) {
//                                        extraMap.put("device_account", imei);
//                                        String i = imei.replace("imei-", "");
//                                        extraMap.put("currentImei", i);
//                                        extraMap.put("video_imei", i);
//                                    } else {
//                                        extraMap.put("device_account", "imei-" + imei);
//                                        extraMap.put("currentImei", imei);
//                                        extraMap.put("video_imei", imei);
//                                    }
//                                }
//                                extraMap.put("platform", mes.getString("platform"));
//                                extraMap.put("video_id", mes.getString("video_id"));
//                                extraMap.put("call_type", mes.getString("video_type"));
//                                extraMap.put("wait_time", mes.getString("wait_time"));
//                                extraMap.put("limit_time", mes.getString("limit_time"));
//                                extraMap.put("user_account", mes.getString("username"));
//                                extraMap.put("dis_width", mes.getString("distinguishability_width"));
//                                extraMap.put("dis_height", mes.getString("distinguishability_heiger"));
//                                extraMap.put("appkey", mes.getString("appkey"));
//
//                                mesCopy.remove("video_type");
//                                mesCopy.remove("wait_time");
//                                mesCopy.remove("distinguishability_heiger");
//                                mesCopy.remove("distinguishability_width");
//
//                                mesCopy.remove("video_id");
//                                mesCopy.remove("username");
//                                mesCopy.remove("distinguishability_width");
//                                mesCopy.remove("distinguishability_heiger");
//                                mesCopy.remove("appkey");
//                                mesCopy.remove("password");
//                                mesCopy.remove("limit_time");
//                                mesCopy.remove("platform");
//                                mesCopy.remove("device_username");
//                                mesCopy.remove("uuid");
//                                mesCopy.remove("wait_time");
                                break;

                            case "videoGroup":
                                isVideoChat = true;
                                vo.setContentAvailable(true);
                                 vo.setVideo(1);
                                vo.setOpenid(mes.getString(ConstPush.OPENID));

                                // 获取发起者名称
                                String initiatorName = mes.getString("initiator");
                                if (StringUtil.isBlank(initiatorName)) {
                                    initiatorName = mes.getString(ConstPush.NAME);
                                }

                                // 设置群视频通话的提醒消息
                                Integer callType = mes.getInteger("call_type");
                                if (callType != null && callType == 1) {
                                    vo.setAlert(initiatorName + "邀请您进行群语音通话");
                                } else {
                                    vo.setAlert(initiatorName + "邀请您进行群视频通话");
                                }

                                vo.setMobType(2);

                                // 设置群视频相关参数
//                                String groupId = mes.getString("groupId");
//                                if (StringUtil.isNotBlank(groupId)) {
//                                    extraMap.put("groupId", groupId);
//                                }
//
//                                // 设置是否群聊标识
//                                extraMap.put("isGroup", "1");
//
//                                // 设置通话类型
//                                extraMap.put("call_type", String.valueOf(callType != null ? callType : 0));
//
//                                // 设置视频房间ID
//                                String videoId = mes.getString("video_id");
//                                if (StringUtil.isNotBlank(videoId)) {
//                                    extraMap.put("video_id", videoId);
//                                }
//
//                                // 设置视频标识
//                                String videoIdentify = mes.getString("video_identify");
//                                if (StringUtil.isNotBlank(videoIdentify)) {
//                                    extraMap.put("video_identify", videoIdentify);
//                                }
//
//                                // 设置其他必要参数
//                                String imei1 = mes.getString(ConstPush.IMEI);
//                                if (imei1 != null) {
//                                    if (imei1.contains("imei-")) {
//                                        extraMap.put("device_account", imei1);
//                                        String i = imei1.replace("imei-", "");
//                                        extraMap.put("currentImei", i);
//                                        extraMap.put("video_imei", i);
//                                    } else {
//                                        extraMap.put("device_account", "imei-" + imei1);
//                                        extraMap.put("currentImei", imei1);
//                                        extraMap.put("video_imei", imei1);
//                                    }
//                                }
//
//                                // 设置平台信息
//                                extraMap.put("platform", mes.getString("platform"));
//
//                                // 设置等待时间和限制时间
//                                extraMap.put("wait_time", mes.getString("wait_time"));
//                                extraMap.put("limit_time", mes.getString("limit_time"));
//
//                                // 设置用户账号
//                                extraMap.put("user_account", mes.getString("username"));
//
//                                // 设置分辨率
//                                extraMap.put("dis_width", mes.getString("distinguishability_width"));
//                                extraMap.put("dis_height", mes.getString("distinguishability_heiger"));
//
//                                // 移除不需要推送的字段
////                                mesCopy.remove("video_id");
////                                mesCopy.remove("video_identify");
//                                mesCopy.remove("username");
//                                mesCopy.remove("distinguishability_width");
//                                mesCopy.remove("distinguishability_heiger");
//                                mesCopy.remove("appkey");
//                                mesCopy.remove("password");
//                                mesCopy.remove("limit_time");
//                                mesCopy.remove("platform");
//                                mesCopy.remove("device_username");
//                                mesCopy.remove("uuid");
//                                mesCopy.remove("wait_time");
////                                mesCopy.remove("initiator");
//                                mesCopy.remove("groupId");
//                                mesCopy.remove("isGroup");
//                                mesCopy.remove("call_type");
                                break;

                            case "unbind":
                                vo.setContentAvailable(true);
                                vo.setAlert(String.format("[%s]手表(%s)的管理员取消您关注手表", mes.getString("name"), mes.getString("imei")));
                                vo.setUnbind(1);
                                vo.setOpenid(mes.getString(ConstPush.OPENID));
                                break;
                            case "unfriend":
                                vo.setContentAvailable(true);
                                vo.setUnfriend(1);
                                vo.setAlert(String.format("设备：%s的其他成员解除了您对设备的关注", mes.getString(ConstPush.IMEI)));
                                break;
                            case "abnormalBodyTemperature":
                            case "abnormalHeartRate":
                                vo.setSound("alarm_sound.caf");
                                vo.setAlert(mes.getString("msgTitle"));
                            default:
                                vo.setContentAvailable(true);
                                break;
                        }
                        //如果从API 那边推过来的有 msgTitle 的字段, 用这个字段设置标题
                        if (StringUtil.isNotBlank(mes.getString(ConstPush.MSG_TITLE))) {
                            vo.setAlert(mes.getString(ConstPush.MSG_TITLE));
                        }
                        vo.setType(dataType);
                        mesCopy.remove("timestamp");
                        mesCopy.remove("msgContent");
                        mesCopy.remove("msgTitle");
                        mesCopy.remove("name");
                    } else if (ConstPush.CHAT_GROUP_MESSAGE.equals(data.getString(ConstPush.TYPE)) || ConstPush.SINGLE_CHAT_GROUP_MESSAGE.equals(data.getString(ConstPush.TYPE))) {
                        vo.setAlert("你收到新的微聊消息");
                        //如果从API 那边推过来的有 msgTitle 的字段, 用这个字段设置标题
                        if (StringUtil.isNotBlank(mes.getString(ConstPush.MSG_TITLE))) {
                            vo.setAlert(mes.getString(ConstPush.MSG_TITLE));
                        }
                        vo.setMobType(3);
                        mesCopy.remove("content");
                        mesCopy.remove("timestamp");
                        mesCopy.remove("msgTitle");
                        extraMap.put("groupId", mes.getString(ConstPush.GROUPID));
                    } else if (ConstPush.CHAT_GROUP_REMOVE.equals(data.getString(ConstPush.TYPE))) {
                        vo.setType(data.getString(ConstPush.TYPE));
                        vo.setAlert(String.format("【%s】的%s取消关注手表", data.get("device_name"), data.get("account_name")));
                        if (StringUtil.isNotBlank(mes)) {
                            if (StringUtil.isNotBlank(mes.getString(ConstPush.MSG_TITLE))) {
                                vo.setAlert(mes.getString(ConstPush.MSG_TITLE));
                            }
                            mesCopy.remove("content");
                            mesCopy.remove("timestamp");
                            mesCopy.remove("msgTitle");
                        }
                        extraMap.put("groupId", data.getString(ConstPush.GROUPID));
                    } else if (ConstPush.CHAT_GROUP_DISBAND.equals(data.getString(ConstPush.TYPE))) {
                        vo.setType(data.getString(ConstPush.TYPE));
                        vo.setAlert(String.format("手表【%s】的管理员将手表恢复出厂设置", data.get("name")));
                        if (StringUtil.isNotBlank(mes)) {
                            if (StringUtil.isNotBlank(mes.getString(ConstPush.MSG_TITLE))) {
                                vo.setAlert(mes.getString(ConstPush.MSG_TITLE));
                            }
                            mesCopy.remove("content");
                            mesCopy.remove("timestamp");
                            mesCopy.remove("msgTitle");
                        }
                        extraMap.put("groupId", data.getString(ConstPush.GROUPID));
                    }
                }
                dataCopy.remove("badge");
                if (vo.isEmpty()) {
                    //其他的指令默认是不在通知栏显示的但是得加上content-available eg:定位 deviceNowLocation 推送
                    vo.setContentAvailable(true);
                }
                vo.setSound("default");
                vo.setData(data);
                vo.setAndroidData(dataCopy);

                for (ApnsVO apnsItem : apns) {
                    vo.setBadge(apnsItem.getBadge() == null ? 0 : apnsItem.getBadge());
//                    if (StringUtil.isNotBlank(vo.getOpenid())) {
//                        if (vo.getOpenid().equals(apnsItem.getOpenid())) {
//                            if ("unbind".equals(vo.getType())) {
//                                vo.setBadge(null);
//                            }
//                            // 推送apns
//                            pushApns(apnsItem, vo);
//                        }
//                    } else {
//                        // 推送apns
//                        pushApns(apnsItem, vo);
//                    }
                    if ("unbind".equals(vo.getType())) {
                        vo.setBadge(null);
                    }
                    // 推送apns
                    pushApns(apnsItem, vo);
                }
                log.info("---------------过滤前的推送的mob对象信息------------:" + mobs);
                //不是视频通话，则需要判断客户端是否在线
                if (!isVideoChat) {
                    for (int m = mobs.size() - 1; m >= 0; m--) {
                        MobVO mob = mobs.get(m);
                        if (isMqttOnLine(mob.getAlias())) {
                            mobs.remove(mob);
                        }
                    }
                }
                log.info("---------------过滤后的推送的mob对象信息------------:" + mobs);
                //异步发送mob
                asyncService.asyncSendMob(mobs, vo, extraMap, data);

            } else {
                log.error("message is null. consumerTopicName={}", consumerTopicName);
            }
        } catch (Exception e) {
            log.error("listen is Exception=" + e.getMessage() + " " + message, e);
        }
    }

    /**
     * 推送给圈子处理
     *
     * @param groupId
     * @param data
     * @param apns
     * @time: 2020/11/27 17:10
     * @author: Rock
     * @version 1.0
     */
    private void chatGroupHandle(String groupId, JSONObject data, List<ApnsVO> apns, List<MobVO> mobs) {
        if (StringUtil.isBlank(groupId)) {
            return;
        }
        JSONObject message = data.getJSONObject("message");
        if (message != null) {
            if (message.containsKey("groupType") && message.getInteger("groupType") == 1) {
                //单聊
                SingleChatGroup chat = singleChatGroupService.findById(groupId);
                if (ObjectUtils.isEmpty(chat)) {
                    log.info("SingleChatGroup is null. groupId={}", groupId);
                    return;
                }
                Account account = accountService.getById(chat.getOpenid());
                if (account == null) {
                    log.info("Account is null. openId={}", chat.getOpenid());
                    return;
                }
                int singleChat = ObjectUtils.isEmpty(chat.getBage()) ? 0 : chat.getBage();
                int groupChat = ObjectUtils.isEmpty(account.getBadge()) ? 0 : account.getBadge();
                //用户未读微聊消息的条数
                data.put(ConstPush.BADGE, singleChat + groupChat);
                // 发送mqtt
                String topic = String.format("iot08/user/%s", chat.getOpenid());
                iMqttSender.sendToMqtt(topic, data.toJSONString());
                log.info("mqtt 推送单聊给用户主题:{}处理 发送成功 openId = {}. data = {}", topic, chat.getOpenid(), data.toJSONString());
                Set<String> openids = new HashSet<>();
                openids.add(chat.getOpenid());
                handlePush(openids, apns, mobs);
            } else {
                //群聊
                ChatGroup chat = chatGroupService.findById(new ObjectId(groupId));
                if (chat != null && CollectionUtil.isNotEmpty(chat.getMemberMap())) {
                    Set<String> keys = chat.getMemberMap().keySet();
                    for (String openId : keys) {
                        //如果发送者的senderID 和 openid 一致, 那么忽略掉它, 不发送给自己
                        //发送者openid
                        String senderOpenid = message.getString("senderID");
                        if (senderOpenid == null) {
                            senderOpenid = message.getString("openid");
                        }
                        if (openId.contains(senderOpenid)) {
                            log.info("发送者的senderID 和 openid 一致, 那么忽略掉它, 不发送给自己 openid:{}", openId);
                            continue;
                        }
                        Account account = accountService.getById(openId);
                        if (account == null) {
                            log.info("Account is null. openId={}", openId);
                            continue;
                        }
                        String imei = message.getString("senderID");
                        String singleGroupId = account.getId() + "&" + imei;
                        SingleChatGroup singleChatGroup = singleChatGroupService.findById(singleGroupId);
                        int badge = account.getBadge() == null ? 0 : account.getBadge();
                        if (!ObjectUtils.isEmpty(singleChatGroup)) {
                            int singleBadge = ObjectUtils.isEmpty(singleChatGroup.getBage()) ? 0 : singleChatGroup.getBage();
                            badge += singleBadge;
                        }
                        data.put(ConstPush.BADGE, badge);
                        // 发送mqtt
                        String topic = String.format("iot08/user/%s", openId);
                        iMqttSender.sendToMqtt(topic, data.toJSONString());
                        log.info("mqtt 群聊推送给圈子里的用户的主题:{} 处理 发送成功 openId = {}. data = {}", topic, openId, data.toJSONString());
                    }
                    handlePush(keys, apns, mobs);
                }
            }
        } else {
            //推群主题
            String topic = String.format("iot08/chatGroup/%s", groupId);
            iMqttSender.sendToMqtt(topic, data.toJSONString());
            log.info("mqtt 群聊推送圈子主题:{} 处理 发送成功  data = {}", topic, data.toJSONString());
            //暂不推个人用户
            //handlePush(keys, apns, mobs);
        }


    }

    /**
     * 推送多用户
     *
     * @param openIds
     * @param data
     * @param apns
     * @time: 2020/11/27 17:09
     * @author: Rock
     * @version 1.0
     */
    private void multipleUsersHandle(List<String> openIds, JSONObject data, List<ApnsVO> apns, List<MobVO> mobs) {
        if (openIds == null || openIds.size() <= 0) {
            return;
        }
        for (String openId : openIds) {
            // 发送mqtt
            iMqttSender.sendToMqtt(String.format("iot08/user/%s", openId), data.toJSONString());
            log.info("mqtt 推送多用户 发送成功 openId = {}. data = {}", openId, data.toJSONString());

        }
        Set<String> ids = new HashSet<>(openIds);
        handlePush(ids, apns, mobs);
    }

    /**
     * 推送单用户
     *
     * @param openId
     * @param data
     * @param apns
     * @time: 2020/11/27 17:09
     * @author: Rock
     * @version 1.0
     */
    private void singleUsersHandle(String openId, JSONObject data, List<ApnsVO> apns, List<MobVO> mobs) {
        if (StringUtil.isBlank(openId)) {
            log.info("send mqtt, openId is null. data={}", data);
            return;
        }
        // 发送mqtt
        iMqttSender.sendToMqtt(String.format("iot08/user/%s", openId), data.toJSONString());
        log.info("mqtt 推送单用户 发送成功 openId = {}. data = {}", openId, data.toJSONString());

        //获取apns处理
        Set<String> ids = new HashSet<>();
        ids.add(openId);
        handlePush(ids, apns, mobs);
    }

    /***
     * 统一判断处理 推送逻辑
     * @param keys
     * @param apns
     * @param mobs
     */
    private void handlePush(Set<String> keys, List<ApnsVO> apns, List<MobVO> mobs) {
        //获取apns处理
        createApnsVOHandle(keys, apns);
        //mob处理
        createMobVOHandle(keys, mobs, apns);

    }


    /**
     * @param topic
     * @param data
     * @time: 2021/3/24 15:53
     * @author: Rock
     * @version 1.0
     */
    private void singleUsersOnlySendMqttHandle(String topic, JSONObject data) {
        if (StringUtil.isBlank(topic)) {
            log.info("send mqtt, topic is null. data={}", data);
            return;
        }
        // 发送mqtt
        iMqttSender.sendToMqtt(topic, data.toJSONString());
        log.info("mqtt 发送成功 topic = {}. data = {}", topic, data.toJSONString());
    }

    /**
     * @param openIds
     * @param mobs
     * @time: 2020/12/1 9:02
     * @author: Rock
     * @version 1.0
     */
    private void createMobVOHandle(Set<String> openIds, List<MobVO> mobs, List<ApnsVO> apns) {
        //根据openId获取对象
        List<OpenidConfig> openids = openidConfigService.listByOpenId(openIds);
        for (OpenidConfig oc : openids) {
            if (!StringUtil.isNumeric(oc.getVersion().replaceAll("\\.", ""))) {
                //不是 1.0.1 这种的直接跳过
                continue;
            }
            if (StringTool.isNewVersionToPush(oc.getPackageName(), oc.getMobileType(), oc.getVersion())) {
                //&& !isMqttOnLine(oc.getOpenid())) {
                log.info("oc.getPackageName():{}, oc.getMobileType(): {}", oc.getPackageName(), oc.getMobileType());
                MobVO mob = new MobVO();
                mob.setPackageName(oc.getPackageName());
                mob.setAlias(oc.getId());
                mob.setRegistrationID(oc.getRegistrationID());
                mob.setOldRegistrationID(oc.getOldRegistrationID());
                if (oc.getMobileType() != null && oc.getMobileType().contains("ios")) {
                    log.info("=================MobileType=="+oc.getMobileType());
                    mob.setPlatform(2);
                    mob.setIosProduction(oc.getIosProduction());
                } else {
                    log.info("=================MobileType=="+oc.getMobileType());
                    mob.setPlatform(1);
                }
                mobs.add(mob);
            }
        }
    }

    /**
     * @param keys
     * @param apns
     * @time: 2021-10-21 18:49
     * @author: Weichao
     * @version 1.0
     */
    private void createApnsVOHandle(Set<String> keys, List<ApnsVO> apns) {
        //根据openId获取对象
        List<OpenidConfig> openids = openidConfigService.listByOpenId(keys);
        for (OpenidConfig oc : openids) {
            if (oc.getMobileType() != null && oc.getMobileType().contains("android")) {
                //安卓的跳过
                continue;
            }
            if (StringTool.isNewVersionToPush(oc.getPackageName(), oc.getMobileType(), oc.getVersion())) {
                //新版本不发送apns
                continue;
            }
            if (isMqttOnLine(oc.getOpenid())) {
                //在线也不发送apns
                continue;
            }
            Account account = accountService.getById(oc.getOpenid());
            if (account == null) {
                continue;
            }
            ApnsVO vo = new ApnsVO();
            vo.setBadge(account.getBadge() == null ? 0 : account.getBadge());
            DeviceToken dt = deviceTokenService.getByOpenid(oc.getOpenid());
            if (dt == null) {
                continue;
            }
            vo.setToken(dt.getId());
            vo.setOpenid(oc.getOpenid());
            vo.setIdent(dt.getPackageName());
            vo.setVersion(dt.getVersion());
//            if (StringUtil.isNotBlank(dt.getVoipDevicetoken())) {
//                vo.setVoipDevicetoken(dt.getVoipDevicetoken());
//            }
//            log.info("添加发送APNS的列表 openid(dt.getA()):{}, uuid:{}, dt.getHaveMqtt():{}",dt.getA(), uuid, dt.getHaveMqtt());
            //添加到发送列表
            apns.add(vo);
        }
    }

    /**
     * @time: 2020/11/5 11:56
     * @author: Rock
     * @version 1.0
     */
    private String createAuthorization() {
        if (StringUtil.isNotBlank(mqttHttpApiKey) && StringUtil.isNotBlank(mqttHttpSecretKey)) {
            return Credentials.basic(mqttHttpApiKey, mqttHttpSecretKey);
        } else {
            byte[] encodedAuth = Base64.encodeBase64("admin:public".getBytes(Charset.forName("US-ASCII")));
            String authHeader = "Basic " + new String(encodedAuth);
            return authHeader;
        }
    }

    /**
     * @param openId
     * @time: 2020/11/30 9:21
     * @author: Rock
     * @version 1.0
     */
    private boolean isMqttOnLine(String openId) {
        if (StringUtil.isNotBlank(openId)) {
            //v4 迁移的 v5 https://www.emqx.io/docs/zh/v5.0/deploy/upgrade-from-v4.html#http-api
            String url = String.format("http://%s:18083/api/v5/clients/%s", mqttServer, openId);
            String authorization = createAuthorization();
            String resp = HttpReq.sendGet(url, authorization);
            if (StringUtil.isNotBlank(resp)) {
                JSONObject json = JSON.parseObject(resp);

                if (json != null && json.getJSONArray(ConstPush.DATA) != null && json.getJSONArray(ConstPush.DATA).size() > 0) {
                    //v4
                    JSONObject data = json.getJSONArray(ConstPush.DATA).getJSONObject(0);
                    if (data.getBoolean(ConstPush.CONNECTED) != null && data.getBoolean(ConstPush.CONNECTED)) {
                        log.info("通过EMQ查询在线>>>, openId={}", openId);
                        //在线状态，只推送mqtt
                        return true;
                    }
                } else {
                    //v5
                    if (json.getBoolean(ConstPush.CONNECTED) != null && json.getBoolean(ConstPush.CONNECTED)) {
                        log.info("通过EMQ查询在线>>>, openId={}", openId);
                        //在线状态，只推送mqtt
                        return true;
                    }
                }
            }
            log.info("通过EMQ查询不在线---, openId={} url:{} resp:{} authorization:{}", openId, url, resp, authorization);
        }

        return false;
    }

    /**
     * @param pushToken
     * @param pushParam
     * @time: 2020/11/27 17:39
     * @author: Rock
     * @version 1.0
     */
    private void pushApns(ApnsVO pushToken, PushSendVO pushParam) {
        String token = pushToken.getToken();
        String pushFun = "dev";
        if ("produce".equals(pushToken.getVersion())) {
            pushFun = "dis";
        }
        if ("video".equals(pushParam.getType()) || "unbind".equals(pushParam.getType())
                || "unfriend".equals(pushParam.getType()) || "reject".equals(pushParam.getType())) {
            if (StringUtil.isNotBlank(pushToken.getOpenid()) && StringUtil.isNotBlank(pushToken.getOpenid())) {
                DeviceToken dt = deviceTokenService.getByOpenid(pushToken.getOpenid());
                if (dt == null) {
                    log.error("pushToken = {} ", pushToken);
                } else {
                    if (dt.getMobileType() != null && dt.getMobileType().contains("ios")
                            && StringUtil.isNotBlank(dt.getSysVersion())) {
                        int sysVersion = 0;
                        String sysv = dt.getSysVersion().split("\\.")[0];
                        if (StringUtil.isNumeric(sysv)) {
                            sysVersion = Integer.parseInt(sysv);
                        }
                        int appVersion = 0;
                        if (StringUtil.isNotBlank(dt.getAppVersion())) {
                            String appv = dt.getAppVersion().replace(".", "");
                            if (StringUtil.isNumeric(appv)) {
                                appVersion = Integer.parseInt(appv);
                            }
                        }
                        if (sysVersion >= 10) {
                            if ("com.jianyou.tongxin".equals(dt.getPackageName()) && appVersion >= 133) {
                                log.info("openid:{} 推送 {}需要修改成 可被APP拦截的APNS推送", pushToken.getOpenid(), dt.getPackageName());
                                pushParam.setMutableContent(true);
                                pushToken.setVoipDevicetoken(null);
                            } else if ("com.tbumall.tb".equals(dt.getPackageName()) && appVersion >= 113) {
                                log.info("openid:{} 推送 {}需要修改成 可被APP拦截的APNS推送", pushToken.getOpenid(), dt.getPackageName());
                                pushParam.setMutableContent(true);
                                pushToken.setVoipDevicetoken(null);
                            } else if ("com.jh.watch".equals(dt.getPackageName()) && appVersion >= 103) {
                                log.info("openid:{} 推送 {}需要修改成 可被APP拦截的APNS推送", pushToken.getOpenid(), dt.getPackageName());
                                pushParam.setMutableContent(true);
                                pushToken.setVoipDevicetoken(null);
                            }
                        }

                    }
                }
            }
//            if (StringUtil.isNotBlank(pushToken.getVoipDevicetoken())) {
//                token = pushToken.getVoipDevicetoken();
//                pushFun = "voip_" + pushFun;
//            }
        }
//        if (pushFun.contains("voip")) {
//            iosPushVOIPMsgWithToken(pushToken.getIdent(), token, pushFun, pushParam);
//        } else {
        iosPushNormalMsg(pushToken.getIdent(), token, pushFun, pushParam);
//        }
    }

    /**
     * @param bundleId
     * @param token
     * @param pushFun
     * @param pushParam
     * @time: 2020/9/28 16:10
     * @author: Rock
     * @version 1.0
     */
    private void iosPushNormalMsg(String bundleId, String token, String pushFun, PushSendVO pushParam) {
        ApnsPayloadBuilder payloadBuilder = new ApnsPayloadBuilder();
        payloadBuilder.setAlertBody(pushParam.getAlert());
        payloadBuilder.setBadgeNumber(pushParam.getBadge());
        payloadBuilder.setContentAvailable(pushParam.getContentAvailable());
        payloadBuilder.setMutableContent(pushParam.getMutableContent());
        payloadBuilder.setSound(pushParam.getSound());
        payloadBuilder.addCustomProperty(ConstPush.DATA, pushParam.getData());

        String payload = payloadBuilder.buildWithDefaultMaximumLength();
        JSONObject pushData = JSONObject.parseObject(payload);
        JSONObject dataObject = pushData.getJSONObject(ConstPush.DATA);
        String imeiRoOpenid = "";
        if (dataObject.containsKey(ConstPush.IMEI)) {
            imeiRoOpenid = dataObject.getString(ConstPush.IMEI);
        } else {
            JSONObject messageOB = dataObject.getJSONObject(ConstPush.MESSAGE);
            if (messageOB != null && messageOB.containsKey("senderID")) {
                imeiRoOpenid = messageOB.getString("senderID");
            }
            if (messageOB != null && messageOB.containsKey(ConstPush.IMEI)) {
                imeiRoOpenid = messageOB.getString(ConstPush.IMEI);
            }
        }
        log.info("imeiRoOpenid={}, 推送内容={} ", imeiRoOpenid, payload);
        try {
            this.pushWithFun(bundleId, payload, token, pushFun);
        } catch (Exception e) {
            log.error("pushWithFun Error: " + e.getMessage(), e);
        }
    }

    /**
     * @param bundleId
     * @param token
     * @param pushFun
     * @param pushParam
     * @time: 2020/9/28 16:10
     * @author: Rock
     * @version 1.0
     */
    public void iosPushVOIPMsgWithToken(String bundleId, String token, String pushFun, PushSendVO pushParam) {
//        ApnsPayloadBuilder payloadBuilder = new ApnsPayloadBuilder();
//        payloadBuilder.setAlertBody(pushParam.getAlert());
//        payloadBuilder.setBadgeNumber(pushParam.getBadge());
//        payloadBuilder.setContentAvailable(pushParam.getContentAvailable());
//        payloadBuilder.setSound(pushParam.getSound());
//        payloadBuilder.addCustomProperty(ConstPush.DATA, pushParam.getData());
//        if (StringUtil.isNotBlank(pushParam.getVideo())) {
//            //视频推送
//            payloadBuilder.addCustomProperty("video", 1);
//        } else if (StringUtil.isNotBlank(pushParam.getUnbind())) {
//            //解绑推送
//            payloadBuilder.addCustomProperty("unbind", 1);
//        }
//        String payload = payloadBuilder.buildWithDefaultMaximumLength();
//        JSONObject pushData = JSONObject.parseObject(payload);
//        JSONObject dataObject = pushData.getJSONObject(ConstPush.DATA);
//        String imeiRoOpenid = "";
//        if (dataObject.containsKey(ConstPush.IMEI)) {
//            imeiRoOpenid = dataObject.getString(ConstPush.IMEI);
//        } else {
//            JSONObject messageOB = dataObject.getJSONObject(ConstPush.MESSAGE);
//            if (messageOB != null && messageOB.containsKey("senderID")) {
//                imeiRoOpenid = messageOB.getString("senderID");
//            }
//            if (messageOB != null && messageOB.containsKey(ConstPush.IMEI)) {
//                imeiRoOpenid = messageOB.getString(ConstPush.IMEI);
//            }
//        }
//        log.info("voip imeiRoOpenid: " + imeiRoOpenid + " 推送内容: " + payload);
//        try {
//            this.pushWithFun(bundleId, payload, token, pushFun);
//        } catch (Exception e) {
//            log.error("pushWithFun Error: " + e.getMessage(), e);
//        }
    }

    /**
     * @param bundleId
     * @param CMD
     * @param token
     * @param pushFUN
     * @time: 2020/11/28 9:12
     * @author: Rock
     * @version 1.0
     */
    private void pushWithFun(String bundleId, String CMD, String token, String pushFUN) {
        PushNotificationFuture<SimpleApnsPushNotification, PushNotificationResponse<SimpleApnsPushNotification>> sendNotificationFuture;
        final SimpleApnsPushNotification pushNotification;
        final ApnsClient tempClient;
//        if (pushFUN.equals("voip_dis")) {
//            pushNotification = new SimpleApnsPushNotification(token, bundleId + ".voip", CMD);
//            tempClient = pushClients.get(bundleId + VOIPDIS);
//        } else if (pushFUN.equals("voip_dev")) {
//            pushNotification = new SimpleApnsPushNotification(token, bundleId + ".voip", CMD);
//            tempClient = pushClients.get(bundleId + VOIPDEV);
//        } else

        if (pushFUN.equals("dis")) {
            pushNotification = new SimpleApnsPushNotification(token, bundleId, CMD);
            tempClient = pushClients.get(bundleId + DIS);
        } else {
            pushNotification = new SimpleApnsPushNotification(token, bundleId, CMD);
            tempClient = pushClients.get(bundleId + DEV);
        }
        if (tempClient == null) {
            log.error("找不到对应BundleID:" + bundleId + " 的推送证书配置参数,token:" + token);
            return;
        }
        sendNotificationFuture = tempClient.sendNotification(pushNotification);
        sendNotificationFuture.addListener(new GenericFutureListener<Future<PushNotificationResponse>>() {
            @Override
            public void operationComplete(Future<PushNotificationResponse> future) throws Exception {
                // This will get called when the sever has replied and returns immediately
                if (future.isSuccess()) {
                    PushNotificationResponse response = future.getNow();
                    if (response != null) {
                        String res = response.getRejectionReason();
                        //APNs网关是否接受推送通知。
                        if (!response.isAccepted()) {
                            log.error("APNs网关不接受推送通知 res: " + res);
                            String token = response.getPushNotification().getToken();
                            if (res != null && res.contains("Unregistered")) {
                                //未注册的 devicetoken 可能是账号什么还没退出,然后就被删掉了
                                deviceTokenService.delById(token);
                            }
                            return;
                        }
                        JSONObject pushData = JSONObject.parseObject(response.getPushNotification().getPayload());
                        JSONObject dataObject = pushData.getJSONObject(ConstPush.DATA);
                        String imeiRoOpenid = "";
                        if (dataObject.containsKey(ConstPush.IMEI)) {
                            imeiRoOpenid = dataObject.getString(ConstPush.IMEI);
                        } else {
                            JSONObject messageOB = dataObject.getJSONObject(ConstPush.MESSAGE);
                            if (messageOB != null && messageOB.containsKey("senderID")) {
                                imeiRoOpenid = messageOB.getString("senderID");
                            }
                            if (messageOB != null && messageOB.containsKey(ConstPush.IMEI)) {
                                imeiRoOpenid = messageOB.getString(ConstPush.IMEI);
                            }
                        }
                        if (res == null) {
                            String token = response.getPushNotification().getToken();
                            log.info("发送成功 imeiRoOpenid: " + imeiRoOpenid + " token: " + token + "  " + response.getTokenInvalidationTimestamp());
                        } else if (res.equals("BadDeviceToken")) {
                            log.info("imeiRoOpenid: " + imeiRoOpenid + " 无效的 devicetoken ,指定的设备令牌无效。验证请求是否包含有效令牌以及令牌是否与环境匹配。");
                        } else {
                            log.error("imeiRoOpenid: " + imeiRoOpenid + " error res: " + res);
                        }
                    } else {
                        log.error("response 为空 ");
                    }
                } else {
                    log.error("I/O操作 写入失败... 推送消息: " + pushNotification.toString());
                    log.error("future.cause(): " + future.cause().getMessage(), future.cause());
                    //重新发送 但是成功失败不管
                    tempClient.sendNotification(pushNotification);
                }
            }
        });
    }


}
